package election.platform;

import java.io.ObjectInputStream;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;

import election.api.Get;
import election.api.Operator;
import election.api.Put;


public class DataSpace implements Runnable
{
    Map<String,MyQueue>     datasets = new HashMap<String,MyQueue>();
    ServerSocket listener;
    
    public DataSpace(int port) throws Exception
    {
        listener = new ServerSocket(port);
    }

    public void run()
    {
        while (true) try {
            Socket conn = listener.accept();
            new DataIntaker(datasets,conn).start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    
    public Get<Object> take(String dataset)
    {
        return datasets.get(dataset);
    }

    public void clear(String dataset)
    {
        datasets.remove(dataset);
        
    }
}


class DataIntaker extends Thread
{
    Socket conn;
    ObjectInputStream in;
    Put exit;
    
    
    public DataIntaker(Map<String,MyQueue> datasets, Socket conn) throws Exception
    {
        this.conn = conn;
        in = new ObjectInputStream(conn.getInputStream());
        String dataset = in.readUTF();
        //boolean isReduce = in.readBoolean();
        MyQueue queue = datasets.get(dataset);
        if (queue == null) {
            queue = new BlockQueue();
            datasets.put(dataset, queue);
        }
        exit = queue;
    }
    
    
    public void run()
    {
        try {
            while (true) {
                Object obj = in.readObject();
                System.out.println("Receive new data");
                exit.put(obj);
            }
        } catch (Exception e) {
            //e.printStackTrace();
            exit.put(null);
            System.out.println("Data END");
        }
    }
}




